#include "StdAfx.h"
#include <winsock2.h>
#include <event.h>
//#include <evhttp.h>
#include <stdint.h>
#include <fcntl.h>
#include <event2/event.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/thread.h>
#include <boost/thread.hpp>
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <io.h>
#include <boost/asio.hpp>
#include <boost/asio/streambuf.hpp>
#include <boost/shared_ptr.hpp>
#include "ClientIOManager.h"


#ifdef   FD_SETSIZE 
#undef   FD_SETSIZE 
#endif
#define   FD_SETSIZE   2048  
#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "wsock32.lib")
#pragma comment(lib, "libevent.lib")
#pragma comment(lib, "libevent_core.lib")
#pragma comment(lib, "libevent_extras.lib")





////////////////////stream input logic//////////////
void input_read(struct bufferevent *bev, void *callBackFunction)
{
	static char inputBuffer[8192];// the buffer containing the input data
	static int inputOffset = 0;       // the length of the input data in the buffer

	void (*pCallBackFunction)(Element& element);
	pCallBackFunction = (void ( *)(Element &)) callBackFunction;
	struct evbuffer *input;

	input = bufferevent_get_input(bev);

	size_t buffer_len = evbuffer_get_length(input);


	
	if (buffer_len <= 0)
		return ;
	if(inputOffset+buffer_len>8192)
	{
		int a  = 1;
		a++;
	}
	assert(inputOffset + buffer_len <= 8192);
	evbuffer_remove(input, inputBuffer + inputOffset, buffer_len);
	inputOffset += buffer_len;
	
	Element element;
	int currentOffset = 0;// how many bytes we have read from the buffer
	while(1)
	{
		if(currentOffset == inputOffset)
		{   // all the input data have been dealed with
			inputOffset = 0;
			break;
		}
		assert(currentOffset<inputOffset);
		//analyze the remaining data, read next element
		//analyze if we can achieve reading the size of the next document
		if(currentOffset+ sizeof(Timestamp)+sizeof(Mark)+sizeof(DocumentId)+sizeof(int) > inputOffset)
		{  // we can not achieve, we copy the remaining data to the front of the buffer and quit 
			memcpy(inputBuffer, inputBuffer+currentOffset, inputOffset - currentOffset);
			inputOffset = inputOffset - currentOffset;
			break;
		}
		//we can achieve the size of the next document
		//analyze if the remaining data is a whole element or not
		int documentSize = *(int*) (inputBuffer + currentOffset + sizeof(Timestamp)+sizeof(Mark)+sizeof(DocumentId));
		int elementSize = sizeof(Timestamp) + sizeof(Mark) + sizeof(DocumentId) + documentSize;
		if(currentOffset + elementSize <= inputOffset)
		{ // yes , it is a whole element in the buffer
			
			
			element.timestamp = *(Timestamp*)(inputBuffer+currentOffset);
			currentOffset += TIMESTAMP_SIZE;

			
			element.id = *(DocumentId*)(inputBuffer+currentOffset);
			currentOffset += DOCUMENT_IDENTIFIER_SIZE;

			
			element.mark = *(Mark*)(inputBuffer + currentOffset);
			currentOffset += MARK_SIZE;

		
			int documentSize = *(int*)(inputBuffer + currentOffset);
			char* documentPosition = inputBuffer + currentOffset;
			currentOffset += documentSize;
			assert(currentOffset <= inputOffset);

			Document document (documentPosition);
			element.document = document;
			element.document.getOwned();

			pCallBackFunction(element);// call the callback function
		}
		else
		{ // no , it is not a whole element in the buffer
			memcpy(inputBuffer, inputBuffer+currentOffset, inputOffset - currentOffset);
			inputOffset = inputOffset - currentOffset;
			break;

		}
	
	
	}
	


	
}






void errorcb(struct bufferevent *bev, short error, void *ctx)//error handler
{
	if (error & BEV_EVENT_EOF) {
		/* connection has been closed, do any clean up here */
		/* ... */
	} else if (error & BEV_EVENT_ERROR) {
		/* check errno to see what error occurred */
		/* ... */
	} else if (error & BEV_EVENT_TIMEOUT) {
		/* must be a timeout event handle, handle it */
		/* ... */
	}
	bufferevent_free(bev);
}

void input_accept(evutil_socket_t listener, short event, void *callBackFunction)//listen socket callback function
{

	struct event_base *base = ClientIOManager::getInstance()->base;
	struct sockaddr_storage ss;
	socklen_t slen = sizeof(ss);
	int fd = accept(listener, (struct sockaddr*)&ss, &slen);
	if (fd < 0) {
		perror("accept");
	} else if (fd > FD_SETSIZE) {
		close(fd);
	} else {
		struct bufferevent *bev;  
		evutil_make_socket_nonblocking(fd);
		bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE); 

		bufferevent_setcb(bev, input_read, NULL, errorcb, callBackFunction); 

		bufferevent_setwatermark(bev, EV_READ, 0, PAGE_SIZE); 
		bufferevent_enable(bev, EV_READ|EV_WRITE); 
	}
}

ClientIOManager * ClientIOManager::ioManager = NULL;
ClientIOManager::ClientIOManager(void)
{
	WSADATA wsaData;

	WORD sockVersion = MAKEWORD(2, 0);
	WSAStartup(sockVersion, &wsaData);
	this->base = event_base_new();  
	if (!this->base)
		return; 
	
}



ClientIOManager::~ClientIOManager(void)
{
	WSACleanup();
}
ClientIOManager* ClientIOManager::getInstance(void)
{
	if(ioManager==NULL)
	{
		ioManager = new ClientIOManager();
	}
	return ioManager;
}
void ClientIOManager::createNewThreadToPerfromIO()
{


	//event_base_loop(this->base,EVLOOP_NONBLOCK);
	event_base_dispatch(this->base);

}
void ClientIOManager::execute()
{
	boost::thread ioThread(boost::bind(&ClientIOManager::createNewThreadToPerfromIO,this));
	ioThread.join();
}

void ClientIOManager::addStreamInput(std::string ip, std::string port, void (*callBackFunction)(Element& element))
{


	int iport = atoi(port.c_str());
	evutil_socket_t listener;
	struct sockaddr_in my_addr;   
	struct event *listener_event;
	my_addr.sin_family = AF_INET;
	my_addr.sin_port = htons(iport);
	my_addr.sin_addr.s_addr = INADDR_ANY;
	listener = socket(AF_INET, SOCK_STREAM, 0 );
	evutil_make_socket_nonblocking(listener);
#ifndef WIN32
	{
		int one = 1;
		setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
	}
#endif
	if(::bind(listener, (struct sockaddr*)&my_addr, sizeof(my_addr))==SOCKET_ERROR)
	{
		perror("bind");
		return;
	};

	if (listen(listener, 16)<0) {  
		perror("listen");
		return;
	}


	listener_event = event_new(base, listener, EV_READ|EV_PERSIST, input_accept, (void*)callBackFunction);  
	/*XXX check it */
	event_add(listener_event, NULL); 

}

